package net.wangds.threadpool.comp;

import net.wangds.log.helper.LogHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.convert.converter.Converter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.validation.constraints.NotNull;
import java.util.List;
import java.util.Stack;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

@Service
public class ThreadPoolComp {

    @Autowired
    @Qualifier("pool.default")
    public ThreadPoolTaskExecutor poolDefault;

    @Autowired
    @Qualifier("pool.msg")
    public ThreadPoolTaskExecutor poolMsg;


    public void runInDefaultPool(Runnable runnable){
        if(runnable!=null){
            poolDefault.execute(runnable);
        }
    }

    public <T> Future<T> runWithFuture(Callable<T> callable){
        return poolDefault.submit(callable);
    }

    public void runWithNotifyFunc(ThreadPoolTaskExecutor pool, Consumer<Runnable> task){
        if(task==null){
            return;
        }
        Object lock = new Object();
        try{
            pool.execute(()-> task.accept(()-> notifyLockObject(lock)));
        }catch(Exception e){
            LogHelper.warn(e);
            notifyLockObject(lock);
        }
    }

    public void runInDefaultPoolWithNotifyFunc(Consumer<Runnable> task){
        runWithNotifyFunc(poolDefault, task);
    }

    public void runInMsgPoolWithNotifyFunc(Consumer<Runnable> task){
        runWithNotifyFunc(poolMsg, task);
    }

    public void runInMsgPool(Runnable runnable){
        if(runnable!=null){
            poolMsg.execute(runnable);
        }
    }

    public void waitOnLockObject(@NotNull Object lock){
        synchronized (lock) {
            try {
                lock.wait();
            } catch (Exception e) {
                LogHelper.error(e);
            }
        }
    }


    public void notifyLockObject(@NotNull Object lock){
        synchronized (lock) {
            try {
                lock.notify();
            } catch (Exception e) {
                LogHelper.error(e);
            }
        }
    }


    public void notifyAllLockObject(@NotNull Object lock){
        synchronized (lock) {
            try {
                lock.notifyAll();
            } catch (Exception e) {
                LogHelper.error(e);
            }
        }
    }

    public void sleep(long millis){
        try{
            Thread.sleep(millis);
        }catch (Exception e){
            LogHelper.error(e);
        }
    }

    public <T> void runBatch(List<T> info, Converter<T, Runnable> conv, int batchSize, Consumer<Throwable> onError){
        if(info==null){
            return;
        }

        runBatch(info.stream().map(conv::convert).collect(Collectors.toList()), batchSize, onError);

    }

    public void runBatch(List<Runnable> tasks, int batchSize, Consumer<Throwable> onError){
        if(tasks==null){
            return;
        }
        Stack<Runnable> st = new Stack<>();
        st.addAll(tasks);
        runBatch(st::pop, batchSize, onError);
    }

    public void runBatch(Supplier<Runnable> taskSupplier, int batchSize, Consumer<Throwable> onError){
        if(taskSupplier==null){
            return;
        }
        batchSize = batchSize<=0?1:batchSize;
        //List<Thread> ths = new ArrayList<>(batchSize);

        for(int i=0;i<batchSize;i++){
            runInDefaultPool(()->{
                while(true){
                    Runnable task = null;
                    try {
                        task = taskSupplier.get();
                    }catch (Throwable th){
                        break;
                    }
                    if(task==null){
                        break;
                    }
                    try{
                        task.run();
                        Thread.sleep(0);
                    }catch(Throwable t){
                        if(onError!=null){
                            onError.accept(t);
                        }else{
                            LogHelper.error(t);
                        }
                    }
                }
            });
        }

        //ths.forEach(Thread::start);

    }

}
